package com.example.traffic.counter.mapreduce;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.PrefixFilter;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.RowCounter;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.MurmurHash;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.NullOutputFormat;

import java.io.IOException;
import java.nio.file.Paths;

public class TrafficCounterApplication {

    private static final String deviceId = "LC06S54L1H1000249";

    public static long size = 0;



    static class OriginDataMapper extends TableMapper<ImmutableBytesWritable, LongWritable> {

        private long r = 0;

        public static LongWritable one = new LongWritable(1);
        @Override
        public void map(ImmutableBytesWritable row, Result values, Context context) throws IOException {

//            String deviceId = new String(row.get());
//            ImmutableBytesWritable userKey = new ImmutableBytesWritable(row.get(), 0, 23);
//            LongWritable v = new LongWritable(values.getValue("data".getBytes(), null).length);

            size += values.getValue("data".getBytes(), null).length;

//            try {
//                context.write(userKey, v);
//            } catch (InterruptedException e) {
//                throw new IOException(e);
//            }
//            if(r++ % 1000 == 0) {
//                System.out.println("r == " + r);
//            }


        }
    }

    static class TrafficCounterReducer extends TableReducer<ImmutableBytesWritable, LongWritable, ImmutableBytesWritable> {

        private long r = 0;
        public void reduce(ImmutableBytesWritable key, Iterable<LongWritable> values, Context context)
                throws IOException, InterruptedException {
            long sum = 0;
            for (LongWritable val : values) {
                sum += val.get();r++;
            }
//            Put put = new Put(key.get());
//            put.addColumn(Bytes.toBytes("details"), Bytes.toBytes("total"), Bytes.toBytes(sum));
            System.out.println(String.format("stats :   key : %s,  count : %d, r: %d", Bytes.toString(key.get()), sum, r));
//            try {
//                context.write(key, put);
//            } catch (Throwable e) {
//                System.out.println(e.getMessage());
//            }




        }
    }



    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {

        HBaseConfiguration conf = new HBaseConfiguration();
        conf.set("hbase.zookeeper.quorum", "10.94.44.91,10.94.44.92,10.94.44.93");
        conf.set("hbase.zookeeper.property.clientPort", "2181");
        Job job = new Job(conf, "Traffic_Counter");
        job.setJarByClass(TrafficCounterApplication.class);
        job.setNumReduceTasks(10);

        Scan scan = new Scan();
        scan.setCacheBlocks(false);
        scan.setCaching(500);
        scan.setStartRow(getDevicePrefix2(deviceId, 9999999999999L).getBytes());
        scan.setStopRow(getDevicePrefix2(deviceId, 0L).getBytes());

        String columns = "data"; // comma seperated
        scan.addColumn(columns.getBytes(), null);

        TableMapReduceUtil.initTableMapperJob("originalData", scan, OriginDataMapper.class, ImmutableBytesWritable.class,
                LongWritable.class, job);

        job.setOutputFormatClass(NullOutputFormat.class);

//        TableMapReduceUtil.initTableReducerJob("summary_user2", TrafficCounterReducer.class, job);

        long s = System.currentTimeMillis();
        job.waitForCompletion(true);
        System.out.println(System.currentTimeMillis() - s);

        System.out.println("size == " + size);

        s = System.currentTimeMillis();
        read();
        System.out.println(System.currentTimeMillis() - s);

    }

    public static Connection createConnection() throws IOException {
        Configuration configuration = HBaseConfiguration.create();
        configuration.set("hbase.zookeeper.quorum", "10.94.44.91,10.94.44.92,10.94.44.93");
        configuration.set("hbase.zookeeper.property.clientPort", "2181");
        return ConnectionFactory.createConnection(configuration);
    }

    private static String getDevicePrefix2(String deviceId, long time) {
        byte[] bytes = deviceId.getBytes();
        int hashCode = MurmurHash.getInstance().hash(deviceId.getBytes());
        if(hashCode < 0) hashCode = 0 - hashCode;
        return String.format("%02d_%20s_%013d", hashCode % 100, deviceId,  Long.MAX_VALUE-time).replace(' ', '0');
    }

    private static Scan setTimeRange(Scan scan, String deviceId, long start, long stop) {
        scan.setStartRow(getDevicePrefix2(deviceId, stop).getBytes());
        scan.setStopRow(getDevicePrefix2(deviceId, start).getBytes());
        return scan;
    }

    public static void read() throws IOException {
        long r = 0;
        try(Connection c = createConnection();
        Table table = c.getTable(TableName.valueOf("originalData"));) {
            Scan scan = new Scan();
            scan.setCacheBlocks(false);
            scan.setCaching(500);
            scan.addColumn("data".getBytes(), null);
            setTimeRange(scan, deviceId, 0, 9999999999999L);

            try(ResultScanner scanner = table.getScanner(scan)) {
                // Reading values from scan result
                for (Result result = scanner.next(); result != null; result = scanner.next()) {
                    r += result.getValue("data".getBytes(), null).length;
                }
            }
        }
        System.out.println("r == "+ r);
    }

}
